package com.example.reactor;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;

public class FluxDemo {

    public static void main(String[] args) {
        /**
         * 发布者
         */
        Flux<Integer> Publisher = Flux.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        /**
         * 订阅者
         */
        Subscriber<Integer> subscriber = new Subscriber<Integer>() {
            /**
             * 订阅关系
             */
            private Subscription s;

            @Override
            public void onSubscribe(Subscription s) {
                System.out.println("订阅关系：" + s);
                this.s = s;
                this.s.request(1);
            }

            @Override
            public void onNext(Integer integer) {
                System.out.println("订阅数据：" + integer);
                this.s.request(1);
                if (integer == 5) {
                    this.s.cancel();
                }
            }

            @Override
            public void onError(Throwable t) {
                System.out.println("订阅数据异常：" + t);
            }

            @Override
            public void onComplete() {
                System.out.println("订阅数据完成");
            }
        };
        /**
         * 发布者绑定订阅者：只有绑定了订阅者，才会有数据发布
         */
        Publisher.subscribe(subscriber);
        /**
         * 休眠5秒，等待数据处理完成
         */
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            System.out.println(e.getMessage());
        }
    }

}
